#ifndef __M_CONSUME_H__
#define __M_CONSUME_H__

#include "../mqcommon/message.pb.h"
#include "../mqcommon/logger.hpp"

#include <iostream>
#include <unordered_map>
#include <mutex>
#include <vector>
#include <memory>
#include <functional>

namespace zhong
{
    // 参数：消费者标识tag 消息属性bp 消息内容body
    using ConsumerCallback = std::function<void(const std::string &, const zhong::BasicProperties *, const std::string &)>;
    struct Consumer
    {
        using ptr = std::shared_ptr<Consumer>;

        std::string tag;           // 消费者标识
        std::string queue_name;    // 订阅的队列名称
        bool auto_ack;             // 自动确认标志
        ConsumerCallback callback; // 消费处理回调函数

        Consumer() {}
        Consumer(const std::string &tag, const std::string &queue_name, bool auto_ac, const ConsumerCallback &cb)
            : tag(tag), queue_name(queue_name), auto_ack(auto_ack), callback(cb) {}
    };
    // 队列单元消费者管理结构
    class QueueConsumer
    {
    public:
        using ptr = std::shared_ptr<QueueConsumer>;

        QueueConsumer(const std::string &queue_name) : _queue_name(queue_name), _rr_seq(0) {}
        // 构建消费者
        Consumer::ptr create(const std::string &tag, const std::string &queue_name, bool auto_ack, const ConsumerCallback &cb)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            // 是否重复
            for (auto &consumer : _consumers)
            {
                if (consumer->tag == tag)
                {
                    ERR_LOG("消费者[%s]已存在，无法重复创建", tag.c_str());
                    return Consumer::ptr();
                }
            }
            // 构造对象
            Consumer::ptr consumer = std::make_shared<Consumer>(tag, queue_name, auto_ack, cb);
            _consumers.push_back(consumer);
            return consumer;
        }
        // 移除消费者
        void remove(const std::string &tag)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            for (auto it = _consumers.begin(); it != _consumers.end(); ++it)
            {
                if (it->get()->tag == tag) // (*it)->tag == tag
                {
                    _consumers.erase(it);
                    return;
                }
            }
        }
        // 队列获取消费者，基于RR轮转获取
        Consumer::ptr choose()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            if (_consumers.size() == 0)  // 不能调用empty作为判断会死锁
            {
                return Consumer::ptr();
            }
            // 进行rr轮转
            int index = _rr_seq % _consumers.size();
            _rr_seq++;

            return _consumers[index];
        }
        bool empty()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _consumers.size() == 0;
        }
        bool exist(const std::string &tag)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            for (auto it = _consumers.begin(); it != _consumers.end(); ++it)
            {
                if (it->get()->tag == tag) // (*it)->tag == tag
                {
                    return true;
                }
            }
            return false;
        }
        void clear()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _consumers.clear();
            _rr_seq = 0;
        }

    private:
        std::string _queue_name; // 订阅的队列名称
        std::mutex _mutex;
        uint64_t _rr_seq;                      // 轮转序号
        std::vector<Consumer::ptr> _consumers; // 管理订阅队列的消费者数据结构
    };


    class ConsumerManager
    {
    public:
        using ptr = std::shared_ptr<ConsumerManager>;
        ConsumerManager() {}
        void initQueueConsumer(const std::string &queue_name)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _queue_consumers.find(queue_name);
            if (it != _queue_consumers.end())
            {
                DEBUG_LOG("队列[%s]已存在，不需要重复初始化", queue_name.c_str());
                return;
            }
            QueueConsumer::ptr queue_consumer = std::make_shared<QueueConsumer>(queue_name);
            _queue_consumers.insert(std::make_pair(queue_name, queue_consumer));
        }
        void destoryQueueConsumer(const std::string &queue_name)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _queue_consumers.find(queue_name);
            if (it == _queue_consumers.end())
            {
                DEBUG_LOG("队列[%s]不存在，无法删除", queue_name.c_str());
                return;
            }
            _queue_consumers.erase(it);
        }
        Consumer::ptr create(const std::string &tag, const std::string &queue_name, bool auto_ack, const ConsumerCallback &cb)
        {
            QueueConsumer::ptr queue_consumer;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_consumers.find(queue_name);
                if (it == _queue_consumers.end())
                {
                    DEBUG_LOG("队列[%s]不存在，无法创建消费者", queue_name.c_str());
                    return Consumer::ptr();
                }
                // 找到了队列
                queue_consumer = it->second;
            }
            // 因为QueueConsumer内部有锁，所以涉及QueueConsumer，不在ConsumerManager临界区进行
            return queue_consumer->create(tag, queue_name, auto_ack, cb);
        }
        void remove(const std::string &tag, const std::string &queue_name)
        {
            QueueConsumer::ptr queue_consumer;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_consumers.find(queue_name);
                if (it == _queue_consumers.end())
                {
                    DEBUG_LOG("队列[%s]不存在，无法删除消费者", queue_name.c_str());
                    return;
                }
                // 找到了队列
                queue_consumer = it->second;
            }
            queue_consumer->remove(tag);
        }
        // 从该队列订阅者中轮转选择消费者进行消息消费
        Consumer::ptr choose(const std::string &queue_name)
        {
            QueueConsumer::ptr queue_consumer;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_consumers.find(queue_name);
                if (it == _queue_consumers.end())
                {
                    DEBUG_LOG("队列[%s]不存在，无法轮转选择消费者", queue_name.c_str());
                    return Consumer::ptr();
                }
                // 找到了队列
                queue_consumer = it->second;
            }
            return queue_consumer->choose();
        }
        bool empty(const std::string &queue_name)
        {
            QueueConsumer::ptr queue_consumer;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_consumers.find(queue_name);
                if (it == _queue_consumers.end())
                {
                    DEBUG_LOG("队列[%s]不存在，无法确定是否为空", queue_name.c_str());
                    return false;
                }
                // 找到了队列
                queue_consumer = it->second;
            }
            return queue_consumer->empty();
        }
        bool exist(const std::string &tag, const std::string &queue_name)
        {
            QueueConsumer::ptr queue_consumer;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_consumers.find(queue_name);
                if (it == _queue_consumers.end())
                {
                    DEBUG_LOG("队列[%s]不存在，无法确定是否存在", queue_name.c_str());
                    return false;
                }
                // 找到了队列
                queue_consumer = it->second;
            }
            return queue_consumer->exist(tag);
        }
        void clear()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            // 对于QueueConsumer的释放，当我们释放这个hash容器后，智能指针会帮我们释放
            _queue_consumers.clear();
        }

    private:
        std::mutex _mutex;
        std::unordered_map<std::string, QueueConsumer::ptr> _queue_consumers;
    };
};
#endif